package com.hkbigdata.flink.task;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerConfig;

import static com.hkbigdata.flink.utils.Constants.*;

public class C01_FlinkCDC_to_Kafka {

    public static void main(String[] args) throws Exception {
        // TODO 1. 准备流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2. 开启检查点   Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,
        // 需要从Checkpoint或者Savepoint启动程序
        // 2.1 开启Checkpoint,每隔5秒钟做一次CK  ,并指定CK的一致性语义
        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        // 2.2 设置超时时间为 1 分钟
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        // 2.3 设置两次重启的最小时间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        // 2.4 设置任务关闭的时候保留最后一次 CK 数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        // 2.5 指定从 CK 自动重启策略
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, Time.days(1L), Time.minutes(1L)
//        ));
//        // 2.6 设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage(
//                "hdfs://hadoop102:8020/flinkCDC"
//        );
//        // 2.7 设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        // TODO 3. 创建 Flink-MySQL-CDC 的 Source
        //initial：首次启动时对受监控的数据库表执行初始快照，并继续读取最新的二进制日志。
        //earliest：首次启动时从不对受监控的数据库表执行快照，只需从二进制日志的开头读取即可。这应该谨慎使用，因为它只有在保证二进制日志包含数据库的整个历史记录时才有效。
        //latest：从不在首次启动时对受监控的数据库表执行快照，只需从二进制日志的末尾读取，这意味着只有自连接器启动以来的更改。
        //specificOffset：首次启动时从不对受监控的数据库表执行快照，直接从指定的偏移量读取 binlog。
        //timestamp：首次启动时从不对监控的数据库表执行快照，直接从指定时间戳读取 binlog。使用者将从一开始就遍历二进制日志，并忽略时间戳小于指定时间戳的更改事件。
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(MYSQL_HOST_NAME)
                .port(MYSQL_PORT)
                .databaseList(MYSQL_DATABASE) // set captured database
//                .tableList("db_pss.ord_order") // set captured table
                .tableList(MYSQL_DATABASE+"."+MYSQL_TABLE_ORDER, MYSQL_DATABASE+"."+MYSQL_TABLE_PRODUCT) // set captured table
                .username(MYSQL_USERNAME)
                .password(MYSQL_PASSWORD)
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.latest())
                .build();

        // TODO 4.使用CDC Source从MySQL读取数据
        DataStreamSource<String> mysqlDS =
                env.fromSource(
                        mySqlSource,
                        WatermarkStrategy.noWatermarks(),
                        "MysqlSource");

        mysqlDS.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                // 打印数据
                out.collect(value);
                System.out.println(value);
            }
        });
        // TODO 5.打印输出
//        mysqlDS.print();//

        // TODO 6.执行任务
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                // 指定 kafka 的地址和端口，
                .setBootstrapServers(KAFKA_BOOSTRAP_SERVERS)
                // 指定序列化器：指定Topic名称、具体的序列化
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic(KAFKA_TOPIC_CDC)
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                // 写到kafka的一致性级别： 精准一次、至少一次
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 如果是精准一次，必须设置 事务的前缀
                .setTransactionalIdPrefix("task-process1-")
                // 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "")
                .build();
        mysqlDS.sinkTo(kafkaSink);

        // TODO 7.执行任务
        env.execute();
    }
}